AWSでジョブWorkerを構成するベストプラクティス 〜 Brianの巻

よく訓練されたアップル信者、都元です。ここまで「AWSでジョブWorkerを構成するベストプラクティス」と題して2本の記事を書いてまいりました。SQSの巻では、バックグラウンドジョブの実行にはSQSを使ってスケーラビリティを確保しようという、下図中の中段の話をしました。続いてBeanstalk worker tierの巻では、Workerの実装方法として、図中の下段の話をしました。今回は、上段の話です。

ジョブスケジューリング

ここまでの話では「ジョブは、ユーザからのWebサーバへのリクエストに伴って発生する」という暗黙の前提で進めてきました。具体的には動画のエンコーディングや、少々時間の掛かる集計処理等をイメージして頂いたと思います。

しかし、ジョブというのは「ユーザのリクエスト」に起因して発生するばかりではありません。ユーザからのリクエストに関わらず「定期的な繰り返し」という、スケジュールに基づいた発生があります。例えば日次・月次等の自動集計処理や、定時連絡的なメール送信等が考えられます。

このようなジョブで皆さんが真っ先に思い浮かべるのがcronでしょう。では直感に従って、cronで10分に一度、SQSにメッセージを送るトリガサーバを思い浮かべてください。

トリガサーバの冗長化

ところで、クラウドの売りの一つは「スケーラビリティ（拡張性）」です。これは単純に、HTTPリクエスト処理要求の増加に伴ってWebサーバの数を増やすだけの話ではありません。昨日ご紹介したように、ロードバランサ、キュー、ワーカー、そして（RDSであれば限定的ですが）データベースのように、システム内の各コンポーネントについて、スケールアウトできる構成が期待されます。また「アベイラビリティ（可用性）」の確保のために、各コンポーネントはAZをまたいで冗長化される必要があります。

このcronによるトリガの仕組みは「スケーラビリティ（拡張性）」面において、どうでしょうか。仮にトリガの数と頻度が爆発的に増加しても、スケールアウトによる性能の確保は可能でしょうか。まぁ、1つ1つのcronの処理は小さな負荷ですので、複数のトリガサーバで分担すればどうにかなるかもしれません。が、要するにシャーディングのようなことをするわけですから、簡単にスケールアウトするとは言い切りづらいところがあります。

まぁまぁ、トリガの発火なんて大した負荷ではありませんので、スケーリングが追いつかない程の大量発火はあまり想像できないので、とりあえず良しとしましょう。しかし、では「アベイラビリティ（可用性）」面についてはどうでしょうか。冗長性の確保のため、同じトリガサーバを2つ作った途端、トリガの発火も二重化されることになります。まぁ、SQSの特性上、同じジョブが複数回発火されても良いような作りにしなければならない、というのは以前お話しした通りですが…。それにしてもほぼ必ず2回以上のメッセージが飛んでくる、というのはコレジャナイ感が激しいのではないでしょうか。

以上より、私的には「AWSネイティブな設計を前提にする場合、cronは選択肢から外れる」という結論になりました。まぁ、使うなってことじゃないんですけど、残念な制限事項が残っちゃうことを受け入れる必要がありますね。

AWS上のトリガとして望まれるもの

ではこのようなジョブスケジューリングをAWSネイティブに実装する場合、どうすれば良いのでしょうか。私は「AWSがサービスとして提供してくれる」のが一番良いと考えていますが、残念ながら本稿執筆時点で、このようなサービスはありません。Amazon Elastic Job Schedulerとかがあればなー。cron表記に従って、そのスケジュールでSNS Topicにメッセージを投げてくれるだけでいいんですけど…。

AWSさん、ご検討よろしくお願いします。

Brian

まぁ、無い物ねだりを続けても進歩が無いので。作りました。Brianと呼んでください。

構成図だけでは伝わりにくいと思うのですがとりあえず。複数のAZにまたがって、複数のEC2インスタンスが起動しています。これらのなかではジョブスケジューラが動いています。すなわちこれらはトリガサーバです。しかし同じトリガが複数台によって発火されてしまわないよう、RDSを介して同期を行っています。発火予定時刻の数秒前に、各トリガサーバが早い者勝ちで発火の担当決めを行います。そしてトリガ発火のタイミングで、SNSにメッセージを投げるようになっています。その先にSQSをつないでおけば、ジョブWorkerまでの道はひらけますね。

そもそものトリガの登録や削除は、トリガサーバにREST風のWeb APIがありますので、そこを叩くことによって実現します。トリガ情報はRDSに書き込まれるため、複数台ある中の何れか1つのAPIさえ叩ければ登録は完了します。というわけで、InternalのELB（VPC内からしか叩けないELB）を介して操作します。

と、偉そうに説明してみましたが、この仕組みはJavaのオープンソースライブラリQuartzを使って実装しています。トリガの登録や削除の操作をWeb APIでラップし、トリガの発火時の処理としてSNSへのメッセージ送信を実装しただけです。詳細はQuartzのドキュメントを御覧ください（丸投げ）。

やってみよう

ここではまず、ざっとポイントを抑えましょう。細かくは色々あるので、実際にBrianを本格的に使ってみようとする際には、次節の詳細仕様をきちんと押さえておいてください。

まず、BrianはJava7で実装されたWebアプリケーションです。warファイルはS3に置いてあります。そして、Elastic Beanstalk (Web tier) で動かすことを前提に考えられています。まぁ、その他環境で動かすことももちろんできますが、色々メンドクサイと思います。

前述の通り、ELBはInternalにします。Brianには認証機構が何もないので、外部公開するのは危険です。ヘルスチェックのエンドポイントは/healthcheckです。

BrianはSNSにメッセージをpublishします。なので、publish先のSNS Topicが必要です。作っておいて、そのARNを控えておいてください。

また、publish操作のためのAPIキーはEC2のInstance Profile経由で取得します。つまり、Brianが動くインスタンスにはInstance Profileが必要です。ざっくりとPowerUserでも良いですし、下記のとおり、きっちり締めあげても構いません。Resourceの項には、先ほど控えたARNを記述します。

{
  "Statement": [
    {
      "Sid": "BrianStmt",
      "Effect": "Allow",
      "Action": "sns:Publish",
      "Resource": "arn:aws:sns:ap-northeast-1:XXXXXXXXXXXX:brian-topic"
    }
  ]
}

その他、前述の通り、同期用のMySQLが必要なので、RDSで構築する必要があります。

JavaのシステムプロパティとしてJDBC_CONNECTION_STRING, DB_USERNAME, DB_PASSWORD, BRIAN_TOPIC_ARNの4つの値が必要です。それぞれ、JDBC接続URI ^*1、DBユーザ名、DBパスワード、発火時のメッセージ送信先SNS Topic ARNを指定します。

って素で聞くと嫌になるもんですよね。そういう時のためのCloudFormationです。

brianのデモを起動

パラメータは以下のとおりです。

• VPC: VPC ID（デフォルトVPCのIDを入れましょう）
• VPCCIDR: VPCのCIDRブロック（デフォルトVPCのブロックを入れましょう）
• Subnet1: サブネットIDその1（デフォルトVPCの(ry）
• Subnet2: サブネットIDその2（デフォルトVPCの(ry）
• DBUsername: RDSのマスターユーザ名
• DBPassword: RDSのマスターパスワード
• KeyName: ECのキーペア名。EC2へのSSH接続用。
• BrianVersion: Brianのバージョンです。今のところ0.5で。

あとはデフォルトのままでContinueボタン連打で構いません。構築所要時間は20〜30分 ^*2です。環境内にbastion（踏み台）が立ち上がりますので、まずはそこにSSH接続します。

続いて、bastionのhome直下に、下記のようなJSONファイルを作って、testEvery20sec.jsonとして保存してください。

{
  "triggerName": "testEvery20sec",
  "scheduleType": "cron",
  "cronEx": "*/20 * * * * ?",
  "timeZone": "Universal",
  "jobData": {
    "foo": "bar",
    "baz": "qux"
  }
}

次に、BeanstalkのManagement Consoleを覗いて、ステータスがGreenであることを確認しつつ、beanstalkのURLをひかえておきます。

そのURLを使って、下記のようなコマンドを叩くことによってトリガの登録ができます。上記のJSONをPOSTしているだけですね。

$ curl -v -H "Content-type: application/json" \
    -X POST http://brian-demo-XXXXXXXXXX.elasticbeanstalk.com/triggers/DEFAULT \
    -d @every20sec.json
* About to connect() to brian-demo-XXXXXXXXXX.elasticbeanstalk.com port 80 (#0)
*   Trying 172.31.24.186...
* Connected to brian-demo-XXXXXXXXXX.elasticbeanstalk.com (XX.XX.XX.XX) port 80 (#0)
> POST /triggers/DEFAULT HTTP/1.1
> User-Agent: curl/7.29.0
> Host: brian-demo-XXXXXXXXXX.elasticbeanstalk.com
> Accept: */*
> Content-type: application/json
> Content-Length: 166
>
* upload completely sent off: 166 out of 166 bytes
< HTTP/1.1 201 Created
< Content-Type: application/json;charset=UTF-8
< Date: Mon, 19 May 2014 04:43:54 GMT
< Server: Apache-Coyote/1.1
< transfer-encoding: chunked
< Connection: keep-alive
<
* Connection #0 to host brian-demo-XXXXXXXXXX.elasticbeanstalk.com left intact
{"success":true,"message":null,"content":{"nextFireTime":"2014-05-19T04:44:00Z"}}

JSONでtestEvery20secというトリガ名を指定しています。また、（世界標準時解釈で）20秒に一度発火するトリガーを定義しています。レスポンスとして、次回の発火予定時刻が返っているのが分かると思います。

このPOSTを行った結果、SNSトピックを介して、SQSキューにみるみるメッセージが溜まっていくのが観察できると思います。興味のある方はSNSトピックをemailでsubscribeしてみると、20秒に1回メールが届いて楽しいです。SQSまで届いたメッセージの一つを抜粋・整形して下記に引用しておきます。

{
  "Type" : "Notification",
  "MessageId" : "XXXXXXXX-XXXX-XXXX-XXXX-XXXXXXXXXXXX",
  "TopicArn" : "arn:aws:sns:ap-northeast-1:XXXXXXXXXXXX:brian-demo-BrianTopic-XXXXXXXXXXXXX",
  "Message" : "{\"brianVersion\":\"0.5\",\"fireTime\":\"2014-05-19T04:44:01Z\",\"scheduledFireTime\":\"2014-05-19T04:44:00Z\",\"nextFireTime\":\"2014-05-19T04:44:20Z\",\"refireCount\":0,\"recovering\":false,\"fireInstanceId\":\"i-XXXXXXXX_1400469148294\",\"trigger\":{\"timeZone\":\"Universal\",\"cronExpression\":\"*/20 * * * * ?\",\"group\":\"DEFAULT\",\"name\":\"testEvery20sec\",\"misfireInstruction\":0,\"startTime\":\"2014-05-19T04:43:54Z\"},\"jobData\":{\"baz\":\"qux\",\"foo\":\"bar\"}}",
  "Timestamp" : "2014-05-19T04:44:02.719Z",
  "SignatureVersion" : "1",
  "Signature" : "BfxBLF9vPfhqeEPOjM2xeBYCJdJ7F2T6MDH7W9MAjuY69VEXAMPLE==",
  "SigningCertURL" : "https://sns.ap-northeast-1.amazonaws.com/SimpleNotificationService-e372f8ca30337fdb084e8ac449342c77.pem",
  "UnsubscribeURL" : "https://sns.ap-northeast-1.amazonaws.com/?Action=Unsubscribe&SubscriptionArn=arn:aws:sns:ap-northeast-1:XXXXXXXXXXXX:brian-demo-BrianTopic-XXXXXXXXXXXXX:XXXXXXXX-XXXX-XXXX-XXXX-XXXXXXXXXXXX"
}

このフォーマットはHTTP/HTTPS Notification JSON Formatに定義されたものです。電子署名により、正規のルートから配信されたメッセージであるかどうか検証することも可能です。

で、Messageの値として入れ子のJSONが入っているのでちょっと見づらいのですが。こちらを整形してみると下記の通りです。

{
  "brianVersion": "0.5",
  "fireTime": "2014-05-19T04:44:01Z",
  "scheduledFireTime": "2014-05-19T04:44:00Z",
  "nextFireTime": "2014-05-19T04:44:20Z",
  "refireCount": 0,
  "recovering": false,
  "fireInstanceId": "i-XXXXXXXX_1400469148294",
  "trigger": {
    "timeZone": "Universal",
    "cronExpression": "*/20 * * * * ?",
    "group": "DEFAULT",
    "name": "testEvery20sec",
    "misfireInstruction": 0,
    "startTime": "2014-05-19T04:43:54Z"
  },
  "jobData": {
    "baz": "qux",
    "foo": "bar"
  }
}

fireTimeは実際の発火時刻、scheduledFireTimeは本来発火予定だった時刻、です。僅かなズレがありますが、分散型である以上許容せざるを得ない誤差です。またjobDataの値としては、トリガ作成時に登録したパラメータがそのまま送られてきています。任意のデータを設定して、好きに使って良い部分です。

まとめ

ジョブスケジューラは、AWS上でHA実装が難しい仕組みでした。この仕組みを導入することにより、高い信頼性を持ったジョブスケジューラを容易に構築できます。

というわけで、あらためまして、ジョブWorker三部作はこれにて完結です。

• AWSでジョブWorkerを構成するベストプラクティス 〜 SQSの巻
• AWSでジョブWorkerを構成するベストプラクティス 〜 Beanstalk worker tierの巻
• AWSでジョブWorkerを構成するベストプラクティス 〜 Brianの巻 （これ）

詳細仕様

以下、長いので付録だと思ってください。本気で使う人用の情報です。

ロケーション

アプリケーション自体はcm-public-eb-applicationsバケットのbrian/brian-0.5.warに公開状態で配置してあります。ご自由に。

デモ起動用のCloudFormationテンプレートの内容もあわせてご確認ください。

起動時のシステムプロパティ

Brianは、動作に必要なパラメータをシステムプロパティとして受け取ります。

トリガ操作エンドポイント

一般的なリクエスト・レスポンス

時刻はISO形式（例: 2014-05-08T06:50:33Z）を用いる。

各エンドポイントは、特に明示のない限り、下記のようなJSON形式のレスポンスを返す。

例1

{
  "success": true,
  "message": null,
  "content": ...
}

例2

{
  "success": false,
  "message": "404 - Not Found",
  "content": null
}

【グループ一覧】GET /triggers

request parameter

なし

response code

response body

例

{
  "content": [
    "DEFAULT"
  ],
  "message": null,
  "success": true
}

【グループ内トリガ一覧】GET /triggers/{triggerGroupName}

request parameter

response code

response body

【トリガ作成】POST /triggers/{triggerGroupName}

request paramter

request body

例

{
  "triggerName": "testEvery20sec",
  "description": "Sample trigger request - fire every 20 seconds.",
  "scheduleType": "cron",
  "misfireInstruction": "DO_NOTHING",
  "cronEx": "*/20 * * * * ?",
  "timeZone": "Universal",
  "endAt": "2015-04-11T20:00:00Z",
  "jobData": {
    "batchJobName": "testEvery20sec",
    "batchJobParameters": {
      "foo" : "bar",
      "baz(long)" : "qux"
    }
  }
}

response code

response body

例

{
  "content": {
    "nextFireTime": "2014-05-08T15:16:00Z"
  },
  "message": null,
  "success": true
}

【トリガ更新】PUT /triggers/{triggerGroupName}/{triggerName}

request parameter

request body

POST /triggers/{triggerGroupName} と同様

response code

response body

POST /triggers/{triggerGroupName} と同様

【トリガ情報参照】GET /triggers/{triggerGroupName}/{triggerName}

request parameter

response code

response body

このエンドポイントだけ一般的なresponse bodyではないので注意。

*/20 * * * * ?

例

{
   "group": "DEFAULT",
   "name": "testEvery20sec",
   "description": "Sample trigger request - fire every 20 seconds.",
   "startTime": "2014-05-08T06:50:33Z",
   "nextFireTime": "2014-05-08T06:50:40Z",
   "misfireInstruction": 2,
   "priority": 5,
   "cronEx": {
      "cronExpression": "*/20 * * * * ?",
      "timeZone": "Universal"
   },
   "jobDataMap": {
      "batchJobParameters": {
         "foo": "bar",
         "baz(long)": "qux"
      },
      "batchJobName": "testEvery20sec"
   }
}

【トリガ削除】DELETE /triggers/{triggerGroupName}/{triggerName}

request parameter

response code

response body

発火時メッセージ

例

{
   "brianVersion": "0.5",
   "fireTime": "2014-05-08T06:54:20Z",
   "scheduledFireTime": "2014-05-08T06:54:20Z",
   "nextFireTime": "2014-05-08T06:54:40Z",
   "refireCount": 0,
   "recovering": false,
   "fireInstanceId": "LOCAL_1399532031174",
   "trigger": {
      "timeZone": "Universal",
      "cronExpression": "*/20 * * * * ?",
      "group": "DEFAULT",
      "name": "testEvery20sec",
      "misfireInstruction": 2,
      "description": "Sample trigger request - fire every 20 seconds.",
      "startTime": "2014-05-08T06:54:05Z"
   },
   "jobData": {
      "batchJobParameters": {
         "foo": "bar",
         "baz(long)": "qux"
      },
      "batchJobName": "testEvery20sec"
   }
}

以上

脚注